feat(storage): Implement ObjectStoreStorage::S3 (Supersedes #2257) - #3165

This parses with Url but then slices the raw input using lengths taken from the normalized parsed fields, and the two don't always line up.

Two concrete failures: an uppercase scheme like S3://bucket/key gets sliced as &path[..2] = "S3", which falls through the match to the unsupported-scheme error and rejects a valid URL. And url.host_str() is percent-decoded, so s3://my%2Dbucket/key gives bucket_str = "my-bucket" (9 bytes) while the raw span is 11 bytes — the bucket slice at line 67 returns the wrong bytes and prefix_len is off, which can panic on a char boundary.

I'd match on url.scheme() directly (it's already lowercased) and pull bucket/relative from url.host_str() / url.path().trim_start_matches('/'), returning owned Strings instead of slicing the input — that's what the opendal sibling does. wdyt?

i find using existing code be used again to be fit

This maps 7 of the 16 S3Config fields and silently drops the rest, and the SSE fields are the dangerous ones: when an operator sets s3.sse.type=kms or custom, TryFrom populates the SSE fields on S3Config but nothing here forwards them, so we write data unencrypted even though encryption was explicitly required. AmazonS3Builder exposes with_sse_kms_encryption / with_ssec_encryption, and the opendal sibling maps all three SSE types — I'd mirror that.

The assume-role fields (role_arn, external_id, role_session_name) and disable_ec2_metadata / disable_config_load are also dropped, and object_store has no builder API for those. Silently ignoring them is worse than not supporting them — a role-based config falls through to the credential chain and only fails at first I/O. I'd return a FeatureUnsupported/DataInvalid error listing the unsupported non-default fields rather than dropping them.

Fix the SSE forwarding and error on the fields we can't express, and this one's resolved.

Since this crate is publish = true, these two variant fields become part of the public API the moment it hits crates.io — public-api.txt already records both as pub. store_cache especially is pure implementation detail; exposing Arc<DashMap<...>> as a public field locks the cache structure into semver, so we couldn't later switch to Mutex<HashMap> or add a store abstraction without a breaking change.

I'd wrap the variant data in a struct with private fields and a pub fn new(config) constructor, exposing the config through a getter if callers need it. Better to lock this down before the first publish than after.

WriteMultipart doesn't complete or abort on drop, and there's no Drop impl here, so if an ObjectStoreWriter is dropped without close() — panic unwind, an early ? return, a cancelled future — the uploaded parts are orphaned in the bucket, billed indefinitely and never committed.

I'd add a Drop that best-effort aborts the inner WriteMultipart via take(). Worth flagging that no test will catch this since it only surfaces as leaked S3 state. wdyt?

This collapses every object_store::Error to ErrorKind::Unexpected, so a NotFound coming back from read/metadata/delete is indistinguishable from a network failure without downcasting the source. exists special-cases NotFound itself, but the others don't, and callers rely on ErrorKind::NotFound for control flow like commit-conflict detection and manifest reads.

I'd dispatch on the object_store::Error variant here — NotFound → ErrorKind::NotFound, PermissionDenied → the closest matching kind, else Unexpected — so every caller gets the right kind for free.

The tests here all run against S3Config::default(), and AmazonS3Builder::build() doesn't validate eagerly, so the cache and roundtrip tests pass without ever touching a backend — none of write/read/reader/delete/delete_prefix/delete_stream/metadata is actually exercised. That's false confidence about exactly the paths most likely to break (multipart lifecycle, serial-vs-batch delete, range reads).

The opendal sibling has a localstack-backed test in CI. I'd add a feature/env-gated integration target covering a write+read roundtrip, a range read, and delete_prefix over 10+ objects before the follow-up backends lean on this crate. wdyt?

am not the best at this, so here i willl take ur and ai assistance

serde = { workspace = true } inherits only features = ["rc"], but this crate uses #[derive(Serialize, Deserialize)]. It compiles in-workspace only because typetag/iceberg happen to activate serde/derive through feature unification — a downstream consumer depending on just this crate would hit use of undeclared crate serde_derive.

Since publish = true, I'd declare it explicitly: serde = { workspace = true, features = ["derive"] }.

ObjectStorePath::from always strips trailing slashes, so ends_with('/') is always false and the else branch just re-appends-then-strips — this whole if/else collapses to let prefix = object_path;. It's only correct today because store.list matches on path-segment boundaries anyway.

Same batching point as delete_prefix — this issues one DeleteObject per path capped at 16 in flight, where DeleteObjects takes 1,000 per request. I'd route this through store.delete_stream(paths) too; if we keep the concurrent form, pull the 16 out into a named const.

ObjectMeta::size is already u64 in object_store 0.13, so this cast is a no-op that trips clippy::useless_conversion. Just size: meta.size,.

The Url rework fixed the uppercase-scheme case from last round, nice. The percent-encoded bucket half is still open though: host_str() returns the host still percent-encoded, so s3://my%2Dbucket/... yields bucket my%2Dbucket, which we hand straight to with_bucket_name and also use as the cache key — no real bucket has that name, and s3://my-bucket vs s3://my%2Dbucket split into two cache entries for the same bucket.

I'd decode the host, or reject any host containing % with DataInvalid. The current test asserts the encoded form, so it's pinning the bug rather than the fix.

The Drop-abort from last round is in, but it can't fire in the case it's for.

close() takes the writer out before finish().await, so if finish() errors the WriteMultipart is already consumed by value — by the time Drop runs, self.writer is None and the guard skips the abort. That's exactly the transient-error path where parts have already been flushed to S3 and now leak until a lifecycle rule expires them.

Holding the lower-level Box<dyn MultipartUpload> instead of WriteMultipart lets us abort() on a failed complete() and again in Drop. wdyt?

but holding box<dyn multipartupload> also means we have to write our own WriteMultipart???

This lists the prefix then deletes one object at a time, each awaited before the next — 10k objects is 10k sequential round-trips, and snapshot expiry hits this with large numbers of manifests. ObjectStoreExt::delete_stream (already imported) maps to S3 DeleteObjects at up to 1000 keys per request.

Piping the list stream into it also lets us drop the trailing-slash branch just above, which is a no-op anyway since ObjectStorePath normalizes trailing slashes. Something like store.list(Some(&prefix)).map_ok(|m| m.location)... fed into delete_stream(...).

Thanks for this one suggestion, got to learn something new

The MinIO tests landed — that closes the "nothing exercises a real read/write" ask. They all go through write(bytes) → put() though; the writer() → WriteMultipart path has no coverage, and that's the primary path for streaming Parquet/Avro data files (and where the Drop/abort issues above live).

A single test that writes past the multipart threshold, closes, and reads back — plus one that drops a writer without closing — would cover both the multipart lifecycle and ask #4 end-to-end.

This branch reaches into object_store internals by string, which gives us an Unexpected error path that can never fire and a silent dependency on a private constant across a crate boundary. There's a typed form:

builder = builder.with_config(
    AmazonS3ConfigKey::Encryption(S3EncryptionConfigKey::ServerSideEncryption),
    "AES256",
);

Worth a server_side_encryption = "AES256" unit test too, since nothing exercises this branch today.

server_side_encryption_customer_key_md5 gets populated by TryFrom when s3.sse.md5 is set, but we only forward custom_key here — the MD5 is dropped. Some S3-compatible stores validate the supplied MD5 and reject SSE-C ops without it. Can we check whether with_ssec_encryption computes it for us, and forward it (or log) if not?

One more thing on this Drop: if it runs outside a Tokio context (try_current() fails on a sync drop or during shutdown) the abort is silently skipped. I'd at minimum tracing::warn! in that branch so an orphaned upload isn't invisible, or document that abort-on-drop only holds from an async context.